package com.acoreful.acf2admin.test.rabbitmq;


import org.apache.thrift.TDeserializer;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

/**
 * Consumes AMQP Messages from the "com.berico.notifications"
 * channel, deserializing the message (using the Thrift API),
 * and printing the contents to the console.
 * @author Richard C. (Berico Technologies)
 */
public class MessageConsumer {

  //The Queue we will be consuming
  public static String QUEUE_NAME = "com.berico.notifications";
  //The Host of the RabbitMQ
  public static String HOST = "localhost";
 
  /**
   * Most of this code is a replication of the RabbitMQ tutorial:
   * http://www.rabbitmq.com/tutorials/tutorial-one-java.html
   * @param args Not used.
   */
  public static void main(String... args){
 
    try {
  
      //Create a connection factory
      ConnectionFactory factory = new ConnectionFactory();
      //Set the host to the location of the RabbitMQ server
      factory.setHost(HOST);
      //Open a new connection
      Connection connection = factory.newConnection();
      //Channel is the abstraction for interacting with a queue
      Channel channel = connection.createChannel();
      //It's possible the Queue doesn't already exist (a producer
      //may not have been instantiated yet).  Create the queue if
      //it doesn't exist
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
      //Comfort logging
      System.out.println(
         " [*] Waiting for messages. To exit press CTRL+C");
      //Queueing Consumer is a buffer for messages, 
      //allowing you handle messages asynchronously 
      //(instead of acting on the message immediately).
      QueueingConsumer consumer = new QueueingConsumer(channel);
      //Register the consumer on the channel
      channel.basicConsume(QUEUE_NAME, true, consumer);

      //Loop until the application is terminated by the user
      while (true) {
        
        //Wait for the next message to arrive in the queue
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      
        //Create a empty notification (we will fill this with the
        //contents of the Thrift serialized payload)
        Notification notification = new Notification();
        //Instantiate the Thrift Deserializer
        TDeserializer deserializer = new TDeserializer();
        //Deserialize the body of the AMQP message, setting the state
        //on the empty notification object
        deserializer.deserialize(notification, delivery.getBody());
      
        //Print the notification to the console.
        System.out.println(" [x] Received '" 
                + notification.toString() + "'");
      }
      
    } catch(Exception e){
      //Dump any error to the console
      e.printStackTrace();
    }
  }
}